package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class 草稿 {
    public static void main(String[] args) {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        // 4 楼栋维表 house
        tabEnv.executeSql(
                "CREATE TABLE ods_ower_clean(\n" +
                        "    id INT,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime TIMESTAMP(3),\n" +
                        "    operatorTime TIMESTAMP(3),\n" +
                        "    proc_time as PROCTIME(),\n" +
                        "    WATERMARK FOR createTime AS createTime - INTERVAL '5' DAYS\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris.ower',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );
        TableResult tableResult = tabEnv.executeSql("SELECT * FROM ods_ower_clean");

        tableResult.print();


    }
}